package org.flk;

import com.google.gson.Gson;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.Map;

public class logHandel {
    public static void main(String[] args) throws Exception {

        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //设置数据来源
        DataStream<Map<String,Object>> mapDataStream=env.addSource(new SqlSeverSource()).name("moruiTest");
        //处理数据来源
        DataStream<Map<String,Object>> alert=mapDataStream.filter(f->f.get("ip").toString().indexOf("192.168.2.1")>0).flatMap(new ToJson()).filter(f->f.get("action").toString().equals("deny"));
        //得到结果
        alert.addSink(new mySink()).name("sink输出");
        //字符串部分为任务名称
        env.execute("run flink");
    }
    public static class ToJson implements FlatMapFunction<Map<String,Object>, Map<String,Object>> {
        @Override
        public void flatMap(Map<String, Object> stringObjectMap, Collector<Map<String, Object>> collector) throws Exception {
            String jsonStr= "{"+stringObjectMap.get("msg").toString()+"}";
            jsonStr=jsonStr.replace(" ", ",");
            Gson gson = new Gson();
            Map<String, Object> map = new HashMap<String, Object>();
            map = gson.fromJson(jsonStr, map.getClass());
            map.put("ip",stringObjectMap.get("ip").toString());
            map.put("saveTime",stringObjectMap.get("saveTime").toString());
            map.put("msg",stringObjectMap.get("msg").toString());
            collector.collect(map);
            System.out.println("");
        }
    }

}